package log

import (
	"context"
	"errors"
	"fmt"
	logv1 "gitee.com/xiao_hange/go-admin-api/api/proto/gen/log/v1"
	"gitee.com/xiao_hange/go-admin-pkg/pkg/logger"
	"gitee.com/xiao_hange/go-admin-pkg/pkg/saramax"
	"github.com/IBM/sarama"
	"go-admin/server/internal/domain"
	"strings"
	"sync"
	"time"
)

type InteractiveReadEventGormLogConsumer struct {
	client      sarama.Client
	l           logger.AppLogger
	mutex       sync.RWMutex
	groupIDName string
	topicName   string
	logSvc      logv1.LogServiceClient
}

type ReadGormLogEvent struct {
	Message string
}

func NewInteractiveReadEventGormLogConsumer(client sarama.Client, l logger.AppLogger, logSvc logv1.LogServiceClient) *InteractiveReadEventGormLogConsumer {
	return &InteractiveReadEventGormLogConsumer{
		client:      client,
		l:           l,
		logSvc:      logSvc,
		groupIDName: "Go_Admin_Log_Record",
		topicName:   "log_record",
	}
}

func (k *InteractiveReadEventGormLogConsumer) Start() error {
	cg, err := sarama.NewConsumerGroupFromClient(k.groupIDName, k.client)
	if err != nil {
		k.l.Error("出错了", logger.Error(err))
		return err
	}
	go func() {
		err = cg.Consume(context.Background(), []string{k.topicName}, saramax.NewHandler[ReadGormLogEvent](k.l, k.groupIDName, k.Consumer))
		if err != nil {
			if errors.Is(saramax.KafkaServerErr, err) {
				for i := 1; i < 10; i++ {
					duration := 20 * time.Second
					timeG, _ := k.reconnectConsumerGroup(cg, i)
					k.l.Warn(fmt.Sprintf("重启第 %d 次失败", i))
					if duration.Milliseconds() < timeG {
						i = 1
					}
					time.Sleep(time.Second * 5)
				}
			} else {
				k.l.Error("退出了消费循环异常", logger.Error(err))
			}
		}
	}()
	return nil
}

// Consumer 具体业务消费 对应在 repo 中
func (k *InteractiveReadEventGormLogConsumer) Consumer(msg *sarama.ConsumerMessage, t ReadGormLogEvent) error {
	fields := strings.Fields(t.Message)
	var index string
	if len(fields) > 4 {
		sourceLog := fields[3] //获取log 来源  如 来源于 Gorm  Http  App 等
		switch sourceLog {
		case "Gorm":
			index = domain.GormLogIndex
		case "HTTP请求":
			index = domain.HttpLogIndex
		default:
			index = domain.AppLogIndex
		}
	}

	_, err := k.logSvc.InputLog(context.Background(), &logv1.InputLogRequest{
		Index:   index,
		Message: t.Message,
	})
	if err != nil {
		k.l.Error("KafKa-Log-Record : 存入ES失败", logger.Error(err), logger.String("Message", t.Message))
		return err
	}
	return nil
}
func (k *InteractiveReadEventGormLogConsumer) reconnectConsumerGroup(cg sarama.ConsumerGroup, i int) (int64, error) {
	k.mutex.Lock()
	defer k.mutex.Unlock()
	startTime := time.Now()
	k.l.Warn(fmt.Sprintf("进行第 %d 重新连接", i))
	newCg, err := sarama.NewConsumerGroupFromClient(k.groupIDName, k.client)
	if err != nil {
		return 0, err
	}
	_ = cg.Close()

	// 启动新的消费者组循环
	err = newCg.Consume(context.Background(), []string{k.topicName}, saramax.NewHandler[ReadGormLogEvent](k.l, k.groupIDName, k.Consumer))
	if err != nil {
		return time.Since(startTime).Milliseconds(), err
	}
	return 0, nil
}
